Skip to content

fix: set MQTT sessionPresent from cleanSession flag and stored session state (#6889) - #6894

Open
lll-peanut wants to merge 9 commits into
apache:masterfrom
lll-peanut:fix/mqtt-session-present
Open

fix: set MQTT sessionPresent from cleanSession flag and stored session state (#6889)#6894
lll-peanut wants to merge 9 commits into
apache:masterfrom
lll-peanut:fix/mqtt-session-present

Conversation

@lll-peanut

@lll-peanut lll-peanut commented Aug 5, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

fix: #6889

  • Connect no longer hard-codes sessionPresent(true); it is derived from the cleanSession flag and stored per-client session state
  • add SessionRepository/MqttSession and resume stored subscriptions on reconnect
  • add unit tests covering first connect, reconnect, cleanSession discard, and subscription resume

DISCONNECT is now dispatched by MqttFactory, so clean-session sessions are removed on graceful disconnect; the disconnected channel is also removed from SubscribeRepository while persistent session state is kept for resume.

Make sure that:

  • You have read the contribution guidelines.
  • You submit test cases (unit or integration tests) that back your changes.
  • Your local test passed ./mvnw clean install -Dmaven.javadoc.skip=true.

@Aias00

Aias00 commented Aug 5, 2026

Copy link
Copy Markdown
Contributor

Good fix — the hardcoded sessionPresent(true) was a clear spec violation (MQTT-3.2.2-6/7), and the new derivation !cleanSession && session != null is correct, including the ordering (sessionPresent computed before the cleanSession overwrite). Two things worth addressing:

Subscription resume loses QoS. Connect.java (~L91) calls SubscribeRepository.add(new ArrayList<>(session.getTopics()), Collections.singletonList(ctx.channel())), which goes through the add(List<String>, List<Channel>) overload (SubscribeRepository.java:43) — a topic→channel mapping only. It bypasses the QoS-aware add(Channel, List<MqttTopicSubscription>) overload (L56). MqttSession.topics is a Set<String>, so the original QoS levels (and duplicate subscriptions with differing QoS) are collapsed; resumed subscriptions get an implicit default QoS rather than the client's original. Could MqttSession store List<MqttTopicSubscription> (or Map<String, MqttQoS>) and resume via the QoS-aware overload?

Clean-session state leaks on graceful DISCONNECT (dead-code cleanup). As the PR body notes, Disconnect isn't dispatched by MqttFactory, so the cleanup added in Disconnect.java:42-52 is unreachable today. Consequence: a clean-session client that disconnects gracefully leaves its MqttSession in SessionRepository until the same clientId next connects clean (Connect.java discards it lazily); clients that never reconnect leak indefinitely. Acceptable as a follow-up, but please either wire Disconnect dispatch in this PR or open a tracking issue and reference it in the PR/code so the leak isn't lost.

Minor (pre-existing, now more reachable): SubscribeRepository.add(List,List) (L43-49) has a check-then-act race — get(s) returns getOrDefault(topic, new CopyOnWriteArrayList<>()), so for an absent topic two concurrent resumes can each create a fresh list and one put overwrites the other's channel. Not introduced here, just noting the resume path makes it reachable.

@lll-peanut
fix: preserve MQTT session state on reconnect
cc761ad 
Store MQTT session subscriptions with their QoS and resume them through the QoS-aware subscribe repository path.

Dispatch DISCONNECT in MqttFactory so clean-session state is removed on graceful disconnect, and unregister disconnected channels from subscriptions.

Use computeIfAbsent in SubscribeRepository to avoid concurrent subscribe/resume overwrites.
@lll-peanut

lll-peanut commented Aug 5, 2026

Copy link
Copy Markdown
Contributor Author

Good fix — the hardcoded sessionPresent(true) was a clear spec violation (MQTT-3.2.2-6/7), and the new derivation !cleanSession && session != null is correct, including the ordering (sessionPresent computed before the cleanSession overwrite). Two things worth addressing:

Subscription resume loses QoS. Connect.java (~L91) calls SubscribeRepository.add(new ArrayList<>(session.getTopics()), Collections.singletonList(ctx.channel())), which goes through the add(List<String>, List<Channel>) overload (SubscribeRepository.java:43) — a topic→channel mapping only. It bypasses the QoS-aware add(Channel, List<MqttTopicSubscription>) overload (L56). MqttSession.topics is a Set<String>, so the original QoS levels (and duplicate subscriptions with differing QoS) are collapsed; resumed subscriptions get an implicit default QoS rather than the client's original. Could MqttSession store List<MqttTopicSubscription> (or Map<String, MqttQoS>) and resume via the QoS-aware overload?

Clean-session state leaks on graceful DISCONNECT (dead-code cleanup). As the PR body notes, Disconnect isn't dispatched by MqttFactory, so the cleanup added in Disconnect.java:42-52 is unreachable today. Consequence: a clean-session client that disconnects gracefully leaves its MqttSession in SessionRepository until the same clientId next connects clean (Connect.java discards it lazily); clients that never reconnect leak indefinitely. Acceptable as a follow-up, but please either wire Disconnect dispatch in this PR or open a tracking issue and reference it in the PR/code so the leak isn't lost.

Minor (pre-existing, now more reachable): SubscribeRepository.add(List,List) (L43-49) has a check-then-act race — get(s) returns getOrDefault(topic, new CopyOnWriteArrayList<>()), so for an absent topic two concurrent resumes can each create a fresh list and one put overwrites the other's channel. Not introduced here, just noting the resume path makes it reachable.

Thanks for the review. I addressed the three points:

  1. MqttSession now stores subscriptions as topic -> MqttQoS, and Connect resumes stored subscriptions through the QoS-aware SubscribeRepository.add(Channel,
    List) path. I also added tests for QoS retention and QoS replacement on resubscribe.

  2. DISCONNECT is now dispatched by MqttFactory. Clean-session DISCONNECT removes the stored session, and DISCONNECT also unregisters the current channel from SubscribeRepository
    while keeping persistent session state for later resume.

  3. SubscribeRepository.add now uses computeIfAbsent instead of the previous get/default-list/put check-then-act sequence, including the QoS-aware overload.

I also added a guard return after rejecting unsupported protocol versions to avoid continuing through authentication/session setup after sending the CONNACK rejection.

Aias00
Aias00 previously approved these changes Sep 4, 2026
View reviewed changes

@Aias00 Aias00 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @lll-peanut!

This properly fixes #6889. The session-present handling now follows MQTT-3.2.2-6/7: sessionPresent is only true for a cleanSession=0 connect when the server holds a stored session for the clientId, instead of the old hard-coded sessionPresent(true). The surrounding session semantics look right too:

  • cleanSession=true discards any previously stored session; cleanSession=false creates a persistent one when absent.
  • Resuming stored subscriptions (topic + QoS) onto the new channel via SubscribeRepository matches the spec's expectation that a persistent session's subscriptions survive reconnects.
  • Wiring DISCONNECT through MqttFactory was the missing dispatch — clean-session state is discarded on graceful disconnect while persistent sessions are kept, and the channel is removed from SubscribeRepository in both cases.
  • Subscribe/Unsubscribe now keep the session's topic→QoS map in sync, including QoS replacement on re-subscribe.
  • The computeIfAbsent change in SubscribeRepository.add also fixes a real race: the old get-then-put could NPE or lose channels under concurrent subscribes.
  • Bonus fix: the missing return after refusing an unsupported protocol version previously let CONNECT processing continue — good catch, and the new test pins that behavior down.

Tests cover first connect, reconnect, clean-session discard, subscription resume, QoS retention/replacement, and DISCONNECT dispatch, using Awaitility appropriately for the async repository operations. The single failing check (k8s-examples-http) is unrelated to shenyu-protocol-mqtt.

One note for follow-up (non-blocking): a client that connects with cleanSession=true and then drops the connection without DISCONNECT leaves its clean session in SessionRepository until the next connect replaces it — harmless given connect-time replacement, but worth tracking if session state grows.

LGTM, approving.

@Aias00

Aias00 commented Sep 18, 2026

Copy link
Copy Markdown
Contributor

Heads-up (PMC Aias00): your PR's CI failures are in shared infrastructure checks (k8s-examples-http, e2e-case-logging-rocketmq, build-docker-images) that are currently failing across many unrelated PRs. This looks like a CI-environment issue, not a problem with your change. Please re-run the failed checks; the code change itself appears fine. No code changes needed for these specific failures.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@Aias00 Aias00 Aias00 left review comments

At least 1 approving review is required to merge this pull request.

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

[BUG] MCP tool-call timeout does not cancel the in-flight plugin chain

2 participants

@lll-peanut @Aias00